{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 8 - Speed up Serialization\n",
    "\n",
    "**GOAL:** The goal of this exercise is to illustrate how to speed up serialization by using `ray.put`.\n",
    "\n",
    "### Concepts for this Exercise - ray.put\n",
    "\n",
    "Object IDs can be created in multiple ways.\n",
    "- They are returned by remote function calls.\n",
    "- They are returned by actor method calls.\n",
    "- They are returned by `ray.put`.\n",
    "\n",
    "When an object is passed to `ray.put`, the object is serialized using the Apache Arrow format (see https://arrow.apache.org/ for more information about Arrow) and copied into a shared memory object store. This object will then be available to other workers on the same machine via shared memory. If it is needed by workers on another machine, it will be shipped under the hood.\n",
    "\n",
    "**When objects are passed into a remote function, Ray puts them in the object store under the hood.** That is, if `f` is a remote function, the code\n",
    "\n",
    "```python\n",
    "x = np.zeros(1000)\n",
    "f.remote(x)\n",
    "```\n",
    "\n",
    "is essentially transformed under the hood to\n",
    "\n",
    "```python\n",
    "x = np.zeros(1000)\n",
    "x_id = ray.put(x)\n",
    "f.remote(x_id)\n",
    "```\n",
    "\n",
    "The call to `ray.put` copies the numpy array into the shared-memory object store, from where it can be read by all of the worker processes (without additional copying). However, if you do something like\n",
    "\n",
    "```python\n",
    "for i in range(10):\n",
    "    f.remote(x)\n",
    "```\n",
    "\n",
    "then 10 copies of the array will be placed into the object store. This takes up more memory in the object store than is necessary, and it also takes time to copy the array into the object store over and over. This can be made more efficient by placing the array in the object store only once as follows.\n",
    "\n",
    "```python\n",
    "x_id = ray.put(x)\n",
    "for i in range(10):\n",
    "    f.remote(x_id)\n",
    "```\n",
    "\n",
    "In this exercise, you will speed up the code below and reduce the memory footprint by calling `ray.put` on the neural net weights before passing them into the remote functions.\n",
    "\n",
    "**WARNING:** This exercise requires a lot of memory to run. If this notebook is running within a Docker container, then the docker container must be started with a large shared-memory file system. This can be done by starting the docker container with the `--shm-size` flag."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import pickle\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.init(num_cpus=4, include_webui=False, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define some neural net weights which will be passed into a number of tasks."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "neural_net_weights = {'variable{}'.format(i): np.random.normal(size=2**18)\n",
    "                      for i in range(50)}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Compare the time required to serialize the neural net weights and copy them into the object store using Ray versus the time required to pickle and unpickle the weights. The big win should be with the time required for *deserialization*.\n",
    "\n",
    "Note that when you call `ray.put`, in addition to serializing the object, we are copying it into shared memory where it can be efficiently accessed by other workers on the same machine.\n",
    "\n",
    "**NOTE:** You don't actually have to do anything here other than run the cell below and read the output.\n",
    "\n",
    "**NOTE:** Sometimes `ray.put` can be faster than `pickle.dumps`. This is because `ray.put` leverages multiple threads when serializing large objects. Note that this is not possible with `pickle`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('Ray - serializing')\n",
    "%time x_id = ray.put(neural_net_weights)\n",
    "print('\\nRay - deserializing')\n",
    "%time x_val = ray.get(x_id)\n",
    "\n",
    "print('\\npickle - serializing')\n",
    "%time serialized = pickle.dumps(neural_net_weights)\n",
    "print('\\npickle - deserializing')\n",
    "%time deserialized = pickle.loads(serialized)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define a remote function which uses the neural net weights."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def use_weights(weights, i):\n",
    "    return i"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** In the code below, use `ray.put` to avoid copying the neural net weights to the object store multiple times."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sleep a little to improve the accuracy of the timing measurements below.\n",
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "\n",
    "results = ray.get([use_weights.remote(neural_net_weights, i)\n",
    "                   for i in range(20)])\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert results == list(range(20))\n",
    "assert duration < 1, ('The experiments ran in {} seconds. This is too '\n",
    "                      'slow.'.format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "382.391px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
